package com.yujiahao.bigdata.test

import org.apache.spark.api.java.JavaRDD.fromRDD
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark02_TestSessionAnalysis {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("HotCategoryTop10SessionAnalysis").setMaster("local[*]")
    val sc = new SparkContext(conf)

    // TODO 需求一
    val rdd = sc.textFile("data/user_visit_action.txt")
    rdd.cache()
    val top10 = rdd.flatMap(
      data => {
        val datas = data.split("_")
        if (datas(6) != "-1") {
          // 数据为点击的场合
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          // 数据为下单的场合
          val ids = datas(8).split(",")
          ids.map((_, (0, 1, 0)))
        } else if (datas(10) != "null") {
          // 数据为支付的场合
          val ids = datas(10).split(",")
          ids.map((_, (0, 0, 1)))
        } else {
          Nil
        }
      }
    ).reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    ).sortBy(_._2, false).take(10)


    //TODO  需求二：Top10热门品类中每个品类的Top10活跃Session统计
    //这里是需求二将拿到前十的品类的id获取到
    val top10Ids: Array[String] = top10.map(_._1)

    //1、将数据进行筛选过滤，只保存前十热门品类的数据。
    //1.1、将数据转换为样例类

    val actionRDD: RDD[UserVisitAction] = rdd.map(
      line => {
        val datas: Array[String] = line.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong

        )
      }
    )
    //1.2、将拿到的数据过滤,将不是点击数的数据过滤出去
    val filterRDD: RDD[UserVisitAction] = actionRDD.filter(
      action => {
        if (action.click_category_id != -1) {
          //这里将数据进行判断是否是前十品类中的数据,注意这里必须转成为字符串，因为类型不一致
          top10Ids.contains(action.click_category_id.toString)
        } else {
          false
        }
      }
    )
    //2、将数据进行结构的转换，line=>((品类,session),1)

    //3、将转换结构后的数据进行统计：((品类,session),1)=> ((品类,session),sum)
    val reduceRDD: RDD[((Long, String), Int)] = filterRDD.map(
      action => {
        ((action.click_category_id, action.session_id), 1)
      }
    ).reduceByKey(_ + _)
    //4、将统计结构进行结构的转换：((品类,session),sum) => (品类,(session,sum))
    //5、将转换结构后的数据按照品类进行分组 (品类,(session,sum) =>(品类，iter((session,sum1),(session,sum2)，(session,sum3)))
    val groupRDD: RDD[(Long, Iterable[(String, Int)])] = reduceRDD.map {
      case ((cid, sessionid), sum) => {
        (cid, (sessionid, sum))
      }
    }.groupByKey()
    //6、将分组后的数据进行排序（降序），取前十名
    val result: RDD[(Long, List[(String, Int)])] = groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    )
    // 7. 将结果采集后打印在控制台
    result.collect().foreach(println)


    //TODO 2、关闭Spark环境的连接
    sc.stop()

  }
  //用户访问动作表
  case class UserVisitAction(
                              date: String,//用户点击行为的日期
                              user_id: Long,//用户的ID
                              session_id: String,//Session的ID
                              page_id: Long,//某个页面的ID
                              action_time: String,//动作的时间点
                              search_keyword: String,//用户搜索的关键词
                              click_category_id: Long,//某一个商品品类的ID
                              click_product_id: Long,//某一个商品的ID
                              order_category_ids: String,//一次订单中所有品类的ID集合
                              order_product_ids: String,//一次订单中所有商品的ID集合
                              pay_category_ids: String,//一次支付中所有品类的ID集合
                              pay_product_ids: String,//一次支付中所有商品的ID集合
                              city_id: Long//城市 id
                            )

}
